// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: listener code decoupling
// feature: support test of communication of simple-client
// feature: DServer support simple entity
// feature: provide query interfaces
// feature: develop dispatcher
// feature: discovery support client and server
// ------------------------------------------------------------------

#include <discovery/participant/PDPServerCompatibleListener.hpp>

#include <memory>

#include <elog/Log.hpp>
#include <edds/dds/builtin/topic/ParticipantBuiltinTopicData.hpp>
#include <ertps/rtps/history/ReaderHistory.h>
#include <history/ParticipantListener.h>
#include <edds/rtps/reader/RTPSReader.h>

#include <edds/rtps/builtin/data/ProxyDataConverters.hpp>
#include <discovery/database/DiscoveryParticipantChangeData.hpp>
#include <discovery/endpoint/EDP.h>
#include <discovery/participant/DS/DiscoveryServerPDPEndpoints.hpp>
#include <discovery/participant/PDPServer.hpp>
// #include <rtps/network/utils/external_locators.hpp>
#include <rtps/participant/RTPSParticipantImpl.h>
#include <deps/common/SystemInfo.hpp>

namespace evbs {
namespace ertps {
namespace rtps {

PDPServerCompatibleListener::PDPServerCompatibleListener(PDPServer* in_PDP) : PDPListener(in_PDP) {}

PDPServer* PDPServerCompatibleListener::pdp_server() {
    return static_cast<PDPServer*>(parent_pdp_);
}
bool PDPServerCompatibleListener::isReconstructMsg(vbs::dispatcher::SubstrateReader* sub_reader,
                                                   const CacheChange_t* const change_in) {
    RTPSReader* reader = static_cast<RTPSReader*>(sub_reader);

    if (reader->getGuid().entityId != ENTITYID_SPDP_BUILTIN_RTPSParticipant_COMPATIBLE_READER) {
        return false;
    }

    CacheChange_t* change = (CacheChange_t*)(change_in);

    // DATA(p|Up) should have a unkown instance handle and no key
    if (change->instanceHandle == c_InstanceHandle_Unknown && !this->get_key(change)) {
        logDebug(RTPS_PDP_LISTENER, "Problem getting the key of the change, removing");  //LCOV_EXCL_START
        logDebug(RTPS_PDP_LISTENER, "------------------ PDP COMPATIBEL SERVER LISTENER END ------------------");
        logDebug(RTPS_PDP_LISTENER, "");
        return false;
    }  //LCOV_EXCL_STOP

    // Take GUID from instance handle
    GUID_t guid = iHandle2GUID(change->instanceHandle);

    // Related_sample_identity could be lost in message delivered, so we set as sample_identity
    // An empty related_sample_identity could lead into an empty sample_identity when resending this msg
    if (change->write_params.related_sample_identity() == SampleIdentity::unknown()) {
        change->write_params.related_sample_identity(change->write_params.sample_identity());
    }

    // DATA(p) case
    if (change->kind == ALIVE) {
        // Ignore announcement from own RTPSParticipant
        if (guid == pdp_server()->getRTPSParticipant()->getGuid()) {
            // Observation: It never reaches this point
            logDebug(RTPS_PDP_LISTENER, "Message from own RTPSParticipant, ignoring");
            logDebug(RTPS_PDP_LISTENER, "------------------ PDP SERVER LISTENER END ------------------");
            logDebug(RTPS_PDP_LISTENER, "");
            return false;
        }

        // Deserialize the payload to access the discovery info
        CDRMessage_t msg(change->serializedPayload);
        temp_participant_data_.clear();
        auto participant_data = temp_participant_data_;

        Time_t startTime;
        if (participant_data.readStartTimeFromCDRMessage(&msg, startTime)) {
            // Grant atomic access to PDP inherited proxies database
            std::lock_guard<std::recursive_mutex> lock(*pdp_server()->startTime_mutex);

            // Check if participant proxy already exists (means the DATA(p) brings updated info)
            auto it = pdp_server()->pdata_startTime.find(guid.guidPrefix);
            logDebug(RTPS_PDP_LISTENER, "Check startup time: " << guid);
            if (it != pdp_server()->pdata_startTime.end() && it->second != 0 && it->second != startTime) {
                return true;
            }
        }
    }
    return false;
}

void PDPServerCompatibleListener::onNewCacheChangeAdded(vbs::dispatcher::SubstrateReader* sub_reader,
                                                        const CacheChange_t* const change_in) {
    logDebug(RTPS_PDP_LISTENER, "");
    logDebug(RTPS_PDP_LISTENER, "------------------ PDP SERVER COMPATIBLE LISTENER START ------------------");
    logDebug(RTPS_PDP_LISTENER,
             "-------------------- " << pdp_server()->mp_RTPSParticipant->getGuid() << " --------------------");
    logDebug(RTPS_PDP_LISTENER, "PDP Server Message received: " << change_in->instanceHandle);

    auto endpoints = static_cast<ertps::rtps::DiscoveryServerPDPEndpoints*>(pdp_server()->builtin_endpoints_.get());
    // Get PDP reader history
    auto pdp_history = endpoints->compatible_reader.history_.get();
    // Get PDP reader to release change
    auto pdp_reader = endpoints->compatible_reader.reader_;

    bool routine_should_be_awake = false;

    bool need_notify = false;
    // Create a delete function to clear the data associated with the unique pointer in case the change is not passed
    // to the database.
    auto deleter = [pdp_history](CacheChange_t* p) {
        // Remove change from reader history, returning it to the pool
        pdp_history->remove_change(p);
    };

    // Unique pointer to the change
    std::unique_ptr<CacheChange_t, decltype(deleter)> change((CacheChange_t*)(change_in), deleter);

    // Get GUID of the writer that sent the change
    GUID_t writer_guid = change->writerGUID;
    RTPSReader* reader = static_cast<RTPSReader*>(sub_reader);

    // DATA(p|Up) should have a unkown instance handle and no key
    if (change->instanceHandle == c_InstanceHandle_Unknown && !this->get_key(change.get())) {
        logDebug(RTPS_PDP_LISTENER, "Problem getting the key of the change(COMPATIBLE), removing");  //LCOV_EXCL_START
        logDebug(RTPS_PDP_LISTENER, "------------------ PDP COMPATIBLE SERVER LISTENER END ------------------");
        logDebug(RTPS_PDP_LISTENER, "");
        return;
    }  //LCOV_EXCL_STOP

    // Take GUID from instance handle
    GUID_t guid = iHandle2GUID(change->instanceHandle);

    // Related_sample_identity could be lost in message delivered, so we set as sample_identity
    // An empty related_sample_identity could lead into an empty sample_identity when resending this msg
    if (change->write_params.related_sample_identity() == SampleIdentity::unknown()) {
        change->write_params.related_sample_identity(change->write_params.sample_identity());
    }

    // Reset the internal CacheChange_t union.
    change->writer_info.next = nullptr;
    change->writer_info.previous = nullptr;
    change->writer_info.num_sent_submessages = 0;
    change->isCompatible = true;

    // DATA(p) case
    if (change->kind == ALIVE) {
        // Ignore announcement from own RTPSParticipant
        if (guid == pdp_server()->getRTPSParticipant()->getGuid()) {
            // Observation: It never reaches this point
            logDebug(RTPS_PDP_LISTENER, "Message from own RTPSParticipant(COMPATIBLE), ignoring");
            logDebug(RTPS_PDP_LISTENER, "------------------ PDP COMPATIBLE SERVER LISTENER END ------------------");
            logDebug(RTPS_PDP_LISTENER, "");
            return;
        }

        // Deserialize the payload to access the discovery info
        CDRMessage_t msg(change->serializedPayload);
        temp_participant_data_.clear();
        auto participant_data = temp_participant_data_;

        if (participant_data.readFromCDRMessage(&msg, true, pdp_server()->getRTPSParticipant())) {
            // if (parent_pdp_->getRTPSParticipant()->is_participant_ignored(participant_data.m_guid.guidPrefix))
            // {
            //     return;
            // }
            if (participant_data.m_domain_id == DOMAIN_ID_UNKNOWN) {
                participant_data.m_domain_id =
                    pdp_server()->getRTPSParticipant()->getDomainIdFromePort(change->dst_port);
                logDebug(RTPS_PDP_LISTENER, "Domain ID is unknown, setting to " << participant_data.m_domain_id
                                                                                << " from port " << change->dst_port);
            }

            std::string participant_type_str;
            auto ret = check_server_discovery_conditions(participant_data, participant_type_str);
            if (!ret.first) {
                return;
            }
            logDebug(RTPS_PDP_LISTENER, "Compatible Participant type " << participant_type_str);
            bool is_simple = ret.second;
            bool is_client = true;
            bool is_local = true;
            bool is_mvbs = participant_data.m_VendorId == c_VendorId_mvbs;

            // For dynamic GUIDs without PID_HOSTID_PID_INFO, extract host_id and pid from the GUID.
            if (!participant_data.m_guid.guidPrefix.is_guid_static()) {
                participant_data.m_hostId = participant_data.m_guid.guidPrefix.get_host_id();
                participant_data.m_pid = participant_data.m_guid.guidPrefix.get_process_id();
            }
            // In case a new changes arrives from a local entity, but the ParticipantProxyData already exists
            //  because we know it from other server
            bool was_local = true;
            if (is_local) {
                was_local = pdp_server()->discovery_db().is_participant_local(change->writerGUID.guidPrefix);
            }

            if (!pdp_server()->discovery_db().backup_in_progress()) {
                // Notify the DiscoveryDataBase
                if (pdp_server()->discovery_db().update(
                        change.get(),
                        ddb::DiscoveryParticipantChangeData(
                            participant_data.metatraffic_locators, is_client, is_local, is_simple, is_mvbs, false,
                            participant_type_str == edds::rtps::ParticipantType::SUPER_CLIENT,
                            participant_data.m_domain_id, participant_data.m_participantName,
                            participant_data.m_startTime, participant_data.m_leaseDuration,
                            participant_data.transport_type))) {
                    // Remove change from PDP reader history, but do not return it to the pool. From here on, the
                    // discovery database takes ownership of the CacheChange_t. Henceforth there are no references to
                    // the change. Take change ownership away from the unique pointer, so that its destruction does not
                    // destroy the data
                    pdp_history->remove_change(pdp_history->find_change(change.release()), false);

                    // Ensure processing time for the cache by triggering the Server thread (which process the updates)
                    // The server does not have to postpone the execution of the routine if a change is received, i.e.
                    // the server routine is triggered instantly as the default value of the interval that the server
                    // has to wait is 0.
                    routine_should_be_awake = true;

                    // TODO: when the DiscoveryDataBase allows updating capabilities we can dismissed old PDP processing
                } else {
                    // If the database doesn't take the ownership, then return the CacheChante_t to the pool.
                    pdp_reader->releaseCache(change.release());
                }

            } else {
                // Release the unique pointer, not the change in the pool
                change.release();
            }

            // At this point we can release reader lock.
            reader->getMutex().unlock();

            // Grant atomic access to PDP inherited proxies database
            std::unique_lock<std::recursive_mutex> lock(*pdp_server()->getMutex());

            // Check if participant proxy already exists (means the DATA(p) brings updated info)
            ParticipantProxyData* pdata = nullptr;
            for (ParticipantProxyData* it : pdp_server()->participant_proxies_) {
                if (guid == it->m_guid) {
                    pdata = it;
                    break;
                }
            }

            // Store whether the participant is new or updated
            auto status = (pdata == nullptr) ? ParticipantDiscoveryInfo::DISCOVERED_PARTICIPANT
                                             : ParticipantDiscoveryInfo::CHANGED_QOS_PARTICIPANT;

            // New participant case
            if (status == ParticipantDiscoveryInfo::DISCOVERED_PARTICIPANT) {
                // TODO: pending avoid builtin connections on client info relayed by other server

                logDebug(RTPS_PDP_LISTENER, "Compatible Registering a new participant: " << guid);
                // Create a new participant proxy entry
                pdata = pdp_server()->createParticipantProxyData(participant_data, writer_guid);
                // Realease PDP mutex
                lock.unlock();

                // If the participant is on the same host as the PDP server, add it to the monitored process
                if (pdp_server()->getRTPSParticipant()->getGuid().is_on_same_host_as(
                        vbs::SystemInfo::instance().host_id(), guid, participant_data.m_hostId)) {
                    pdp_server()->add_monitored_process(participant_data.m_pid, guid);
                }

                // All local builtins are connected, the database will avoid any EDP DATA to be send before having PDP
                // DATA acknowledgement. Non-local SERVERs will also be connected
                if (pdata && is_local) {
                    pdp_server()->assignRemoteEndpoints(pdata);
                    need_notify = true;
                }
            }
            // Case ParticipantProxyData already exists but was known remotely and now must be local
            else if (is_local && !was_local) {
                // Realease PDP mutex
                lock.unlock();

                pdp_server()->assignRemoteEndpoints(pdata);
                need_notify = true;
            }
            // Updated participant information case
            else {
                logDebug(RTPS_PDP_LISTENER, "Compatible Updating a new participant: " << guid);
                if (pdata->m_startTime != 0 && pdata->m_startTime != participant_data.m_startTime) {
                    // All local builtins are connected, the database will avoid any EDP DATA to be send before having PDP
                    // DATA acknowledgement. Non-local SERVERs will also be connected
                    if (is_local) {
                        pdp_server()->discovery_db().delete_entities_of_participant(guid);
                        pdp_server()->remove_remote_all_entities(
                            pdata, pdata->m_guid, ParticipantDiscoveryInfo::REMOVED_PARTICIPANT, nullptr);

                        pdp_server()->assignRemoteEndpoints(pdata);
                        need_notify = true;
                        logDebug(RTPS_PDP_LISTENER, "Compatible Reconstruct participant startTime update."
                                                        << "old StartTime: " << pdata->m_startTime.seconds() << "s "
                                                        << pdata->m_startTime.nanosec() << "ns "
                                                        << "new StartTime: " << participant_data.m_startTime.seconds()
                                                        << "s " << participant_data.m_startTime.nanosec() << "ns "
                                                        << "guid:" << guid);
                    }
                }
                // Update proxy
                pdata->updateData(participant_data);
                pdp_server()->update_guid_startTime(participant_data.m_guid.guidPrefix, participant_data.m_startTime);
                pdata->isAlive = true;
                // Realease PDP mutex
                lock.unlock();

                // TODO: pending client liveliness management here
                // Included form symmetry with PDPListener to profit from a future updateInfoMatchesEDP override
                if (pdp_server()->updateInfoMatchesEDP() && is_local) {
                    pdp_server()->mp_EDP->assignRemoteEndpoints(*pdata, true);
                }
            }

            // Check whether the participant proxy data was created/updated correctly
            if (pdata != nullptr && need_notify) {
                // Notify user of the discovery/update of the participant
                vbs::common::ParticipantListener* listener = pdp_server()->getRTPSParticipant()->getListener();
                if (listener != nullptr) {
                    {
                        std::lock_guard<std::mutex> cb_lock(pdp_server()->callback_mtx_);
                        ParticipantDiscoveryInfo info(*pdata);
                        info.status = ParticipantDiscoveryInfo::DISCOVERED_PARTICIPANT;

                        listener->onParticipantDiscovery(pdp_server()->getRTPSParticipant()->getUserRTPSParticipant(),
                                                         std::move(info));
                    }
                }
            }

            // Take again the reader lock
            reader->getMutex().lock();
        }
    }

    /*
     * Awake routine thread if needed.
     * Thread is awaken at the end of the listener as it is required to have created the Proxies before
     * the data is processed and the new messages added to history.
     * If not, could happen that a message is added to history in order to be sent to a relevant participant, and
     * this Participant still not have a ReaderProxy associated, and so it will miss the message and it wont be
     * sent again (because if there are no changes PDP is no sent again).
     */
    if (routine_should_be_awake) {
        pdp_server()->awake_routine_thread();
    }

    // cache is removed from history (if it's still there) and returned to the pool on leaving the scope, since the
    // unique pointer destruction grants it. If the ownership has been taken away from the unique pointer, then nothing
    // happens at this point

    logDebug(RTPS_PDP_LISTENER,
             "-------------------- " << pdp_server()->mp_RTPSParticipant->getGuid() << " --------------------");
    logDebug(RTPS_PDP_LISTENER, "------------------ COMPATIBLE PDP SERVER LISTENER END ------------------");
    logDebug(RTPS_PDP_LISTENER, "");
}

std::pair<bool, bool> PDPServerCompatibleListener::check_server_discovery_conditions(
    const ParticipantProxyData& participant_data, std::string& participant_type_str) {
    // is_valid, is_simple
    std::pair<bool, bool> ret {true, true};

    // In Discovery Server we don't impose
    // domain ids to be the same
    /* Do not check PID_DOMAIN_ID */

    edds::dds::ParameterPropertyList_t properties = participant_data.m_properties;

    /* Check DS_VERSION */
    if (ret.first) {
        auto ds_version =
            std::find_if(properties.begin(), properties.end(), [](const edds::dds::ParameterProperty_t& property) {
                return property.first() == edds::dds::parameter_property_ds_version;
            });

        if (ds_version != properties.end()) {
            if (std::stof(ds_version->second()) < 1.0) {
                elogError(RTPS_PDP_LISTENER, RetCode_t::RETCODE_PRECONDITION_NOT_MET,
                          "Minimum " << edds::dds::parameter_property_ds_version
                                     << " is 1.0, found: " << ds_version->second());
                ret.first = false;
            }
            logDebug(RTPS_PDP_LISTENER, "Compatible Participant " << edds::dds::parameter_property_ds_version << ": "
                                                                  << ds_version->second());
        } else {
            logDebug(RTPS_PDP_LISTENER, edds::dds::parameter_property_ds_version << " is not set. Assuming 1.0");
        }
    }

    /* Check PARTICIPANT_TYPE */
    if (ret.first) {
        participant_type_str = parent_pdp_->check_participant_type(properties);

        if (participant_type_str != edds::rtps::ParticipantType::SIMPLE) {
            logDebug(RTPS_PDP_LISTENER, "Compatible Reader ignoring " << parameter_property_participant_type << ": "
                                                                      << participant_type_str);
            ret.first = false;
        }
        logDebug(RTPS_PDP_LISTENER, "Compatible Participant type " << participant_type_str);
    }

    return ret;
}

} /* namespace rtps */
}  // namespace ertps
} /* namespace evbs */
